package com.message.queue.rocket.defaultmq;

import com.message.queue.constant.RocketConstant;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

/**
 * Created by 张波 on 2019/9/22.
 */
public class DefaultProduct {


    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("default_producer");
        producer.setNamesrvAddr(RocketConstant.NAME_SERVER);
        producer.start();

        for (int i = 0; i < 5; ++i) {
            Message message = new Message("test_quick_topic",  //主题
                    "TagA", //标签
                    "key" + i,  //唯一标识
                    ("Hello Message" + i).getBytes());  //消息内容
            //1、发送同步消息
//            SendResult result = producer.send(message);
            //2、发送异常消息
//            producer.send(message, new SendCallback() {
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    System.out.println("msgId:" + sendResult.getMsgId() + ",sendStatus:" + sendResult.getSendStatus() + ",messageQueue" + sendResult.getMessageQueue());
//                }
//                @Override
//                public void onException(Throwable e) {
//                    e.printStackTrace();
//                }
//            });
            //3、将消息发送到指定队列（可用于同一类消息要求保证顺序时）
            SendResult result = producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    //参数args是从sned方法中最后一个参数中获取，消息队列默认队列数为4个，可通过broker的defaultTopicQueueNums进行配置
                    Integer queueIndex = (Integer) arg;
                    return mqs.get(queueIndex);

                }
            }, 3);
            if (result.getSendStatus().equals(SendStatus.SEND_OK)) {
                //发送成功

            } else {
                //TODO 发送异常分以下三种情况，对于必须保证消息一定不会丢失的场景，必须考虑做消息发送补偿机制，以保证消息不丢失
                /*服务端写磁盘超时*/
//                SendStatus.FLUSH_DISK_TIMEOUT
                /*同步从节点超时*/
//                SendStatus.FLUSH_SLAVE_TIMEOUT
                /*从节点不可用*/
//                SendStatus.SLAVE_NOT_AVAILABLE
            }

            System.out.println("result:" + result);
        }
        //发送异常消息时不能关掉，防止进行回调时发送者已经被关掉了（会报异常）
        producer.shutdown();

    }
}
